package com.ityuwen.mengyagushi.user.service.impl;

import com.alibaba.fastjson.JSON;
import com.ityuwen.mengyagushi.common.constant.KafkaConstant;
import com.ityuwen.mengyagushi.common.constant.RedisConstant;
import com.ityuwen.mengyagushi.common.constant.SystemConstant;
import com.ityuwen.mengyagushi.common.service.KafkaService;
import com.ityuwen.mengyagushi.common.util.AuthContextHolder;
import com.ityuwen.mengyagushi.common.util.MongoUtil;
import com.ityuwen.mengyagushi.model.user.UserListenProcess;
import com.ityuwen.mengyagushi.user.service.UserListenProcessService;
import com.ityuwen.mengyagushi.vo.album.TrackStatMqVo;
import com.ityuwen.mengyagushi.vo.user.UserListenProcessVo;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.temporal.ChronoUnit;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class UserListenProcessServiceImpl implements UserListenProcessService {

	@Autowired
	private MongoTemplate mongoTemplate;

	@Autowired
	private RedisTemplate redisTemplate;

	@Autowired
	private KafkaService kafkaService;

	@Override
	public BigDecimal getTrackBreakSecond(Long trackId) {

		Long userId = AuthContextHolder.getUserId();
		// 用户未登录则从第0秒播放
		if (userId == null){
			return new BigDecimal(0);
		}

		// 如果登陆了则查询播放进度
		String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
		UserListenProcess userListenProcess = this.mongoTemplate.findOne(Query.query(Criteria.where("userId").is(userId).and("trackId").is(trackId)),
				UserListenProcess.class, collectionName);
		// 如果用户的播放进度不存在则直接返回0
		if (userListenProcess == null){
			return new BigDecimal(0);
		}

		return userListenProcess.getBreakSecond();
	}

	@Override
	public void updateListenProcess(UserListenProcessVo userListenProcessVo) {
		// 1.获取登录 信息，判断登录状态
		Long userId = AuthContextHolder.getUserId();
		if (userId == null) {
			return;
		}

		// 2.查询播放进度
		String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
		UserListenProcess userListenProcess = this.mongoTemplate.findOne(Query.query(Criteria.where("userId").is(userId)
						.and("trackId").is(userListenProcessVo.getTrackId())), UserListenProcess.class,
				collectionName);

		// 判断播放进度是否为空
		if (userListenProcess != null) {
			// 3.更新播放进度
			userListenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
			userListenProcess.setUpdateTime(new Date());
		} else {
			// 4.新增播放进度
			userListenProcess = new UserListenProcess();
			BeanUtils.copyProperties(userListenProcessVo, userListenProcess);
			userListenProcess.setId(IdWorker.getIdStr());
			userListenProcess.setUserId(userId);
			userListenProcess.setIsShow(1);
			userListenProcess.setCreateTime(new Date());
			userListenProcess.setUpdateTime(userListenProcess.getCreateTime());
		}
		this.mongoTemplate.save(userListenProcess, collectionName);

		// 5.查询是否统计过：保证四个一（同一个用户同一个声音同一天只统计一次）
		String key = RedisConstant.USER_PROCESS_PREFIX + userId;
		Boolean flag = this.redisTemplate.opsForValue().getBit(key, userListenProcessVo.getTrackId());
		if (flag){
			return;
		}
		// 先记录为已统计
		this.redisTemplate.opsForValue().setBit(key, userListenProcessVo.getTrackId(), true);
		LocalDateTime expireTime = LocalDateTime.of(LocalDate.now().plusDays(1), LocalTime.MIN);
		// 每天凌晨过期
		this.redisTemplate.expire(key, ChronoUnit.SECONDS.between(LocalDateTime.now(), expireTime), TimeUnit.SECONDS);

		try {
			// 6.发送消息给kafka
			TrackStatMqVo statMqVo = new TrackStatMqVo();
			BeanUtils.copyProperties(userListenProcessVo, statMqVo);
			statMqVo.setStatType(SystemConstant.TRACK_STAT_PLAY);
			statMqVo.setCount(1);
			statMqVo.setBusinessNo(IdWorker.getIdStr()); // 防止消息重复消费的唯一标识
			this.kafkaService.sendMsg(KafkaConstant.QUEUE_TRACK_STAT_UPDATE, JSON.toJSONString(statMqVo));
		} catch (BeansException e) {
			this.redisTemplate.opsForValue().setBit(key, userListenProcessVo.getTrackId(), false);
			throw new RuntimeException(e);
		}
	}

	@Override
	public UserListenProcess getLatelyTrack() {
		Long userId = AuthContextHolder.getUserId();
		if (userId == null){
			// TODO: 推广的声音
			return null;
		}
		UserListenProcess userListenProcess = this.mongoTemplate.findOne(Query.query(Criteria.where("userId").is(userId))
						.with(Sort.by(Sort.Order.desc("updateTime"))), UserListenProcess.class,
				MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId));
		if (userListenProcess == null){
			// TODO：推广的声音
			return null;
		}
		return userListenProcess;
	}

	public static void main(String[] args) {
		LocalDateTime now = LocalDateTime.now();
		LocalDateTime endTime = LocalDateTime.of(LocalDate.now().plusDays(1), LocalTime.MIN);
		System.out.println(endTime);
		System.out.println(ChronoUnit.SECONDS.between(now, endTime));
	}
}
